package com.atguigu.day6

import java.sql.{Connection, DriverManager, PreparedStatement}

import com.atguigu.source.{SensorReading, SensorSource}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._

object WriteToMySQL {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val stream = env.addSource(new SensorSource)
    stream.addSink(new MyJdbcSink)
    env.execute()
  }
  class MyJdbcSink extends RichSinkFunction[SensorReading]{
    var conn:Connection = _
    var insertStmt:PreparedStatement = _
    var updataStmt:PreparedStatement = _

    override def open(parameters: Configuration): Unit = {
      conn = DriverManager.getConnection(
        "jdbc:mysql://192.168.35.102:3306/test",
        "root",
        "000000"
      )

      insertStmt = conn.prepareStatement(
        "INSERT INTO temperatures (sensor,temp) VALUE (?,?)"
      )
      updataStmt = conn.prepareStatement(
        "UPDATE temperatures SET temp = ? WHERE sensor = ?"
      )

    }

    override def invoke(value: SensorReading, context: SinkFunction.Context[_]): Unit = {
      updataStmt.setDouble(1,value.temperature)
      updataStmt.setString(2,value.id)
      updataStmt.execute()
      if(updataStmt.getUpdateCount == 0){
        insertStmt.setString(1,value.id)
        insertStmt.setDouble(2,value.temperature)
        insertStmt.execute()
      }

    }

    override def close(): Unit = {
      insertStmt.close()
      updataStmt.close()
      conn.close()
    }


  }
  }
